package cn.com.jonpad;

import org.junit.Test;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

/**
 * @author Jon Chan
 * @date 2019/4/13 23:39
 */
public class FlowDemo {

    Flow.Subscriber<Integer> getSubscriber() {
        return new Flow.Subscriber<>() {

            private Flow.Subscription subscription;

            /**
             * 建立订阅关系时调用
             * @param subscription
             */
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                // 保存订阅关系
                this.subscription = subscription;
                System.out.println("订阅已建立");

                // 首次请求数据
                // request(long n) 参数n 需要多少条 接收n个onNext调用
                this.subscription.request(1L);
            }

            /**
             * 当有数据到时，onNext方法会被触发
             * @param item
             */
            @Override
            public void onNext(Integer item) {
                System.out.println("接收到数据：" + item);
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 关键！接受到数据后继续请求下一条
                this.subscription.request(1L);
                // 如果需要停止可以
                // this.subscription.cancel();

            }

            @Override
            public void onError(Throwable throwable) {
                throwable.printStackTrace();
                this.subscription.cancel();
            }

            /**
             * publisher.close();触发
             */
            @Override
            public void onComplete() {
                System.out.println("完成！");
            }
        };
    }


    /**
     * JDK 11 Flux API示例
     * @throws Exception
     */
    @Test
    public void test() throws Exception {
        // 1.定义发布者
        // SubmissionPublisher类实现了Publisher接口
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

        // 2.定义订阅者

        Flow.Subscriber<Integer> subscriber = getSubscriber();

        // 3.建立关系
        publisher.subscribe(subscriber);

        // 4.生成数据，并且发布
        for (int i = 0; i < 1000; i++) {
            publisher.submit(i);
            System.out.println("生成数据：" + i);
        }

        // 5.关闭发布者
        publisher.close();

        // join()的作用是：“等待该线程终止”，这里需要理解的就是该线程是指的主线程等待子线程的终止。也就是在子线程调用了join()方法后面的代码，只有等到子线程结束了才能执行。
        Thread.currentThread().join(1000);
    }

}
